package com.luoqifei.yomob.realtime.analysis.kafka.stream;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.lqf.common.Constants;
import com.lqf.common.CounterRecord;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KStreamBuilder;
import org.apache.kafka.streams.kstream.KeyValueMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;

import java.util.Properties;

public class DauTopology {
    private static Logger LOG = LoggerFactory.getLogger(DauTopology.class);
    private KStreamBuilder kStreamBuilder = new KStreamBuilder();
    private StreamsConfig streamsConfig;
    private Properties conf = new Properties();
    private Jedis jedis;
    private void init(){}
    public void buildTopology(String sink,String src){
        KStream<String,String> kStream = kStreamBuilder.stream(sink);

        KStream<String,Long> count = kStream.map(new KeyValueMapper<String, String, KeyValue<String, Long>>() {
            @Override
            public KeyValue<String, Long> apply(String key, String value) {
                LOG.debug("get a new message"+"key is "+key+" value is "+value);
                JSONObject jsonObject = JSON.parseObject(value);
                CounterRecord counterRecord = JSON.parseObject(value,CounterRecord.class);
                jedis.set(counterRecord.getAppid(),"1");
                jedis.incr("dauSum_20170901");
                jedis.incr("appId_20170901");
                jedis.incr("appId_20170901_last_1");
                jedis.incr("appId_20170901_last_2");
                jedis.incr("appId_20170901_2");

                return new KeyValue<String,Long>(counterRecord.getAppid(),1L);
            }
        }).through(src);
    }
    public void setJedis(String redisIp, int port){
        jedis = new Jedis(redisIp, port);
    }
    public void setConfig(String kafkaUrl, String zkUrl) {
        conf.put(StreamsConfig.APPLICATION_ID_CONFIG, Constants.DAU_COUNT_PROCESSOR_ID);
        conf.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaUrl);
        conf.put(StreamsConfig.ZOOKEEPER_CONNECT_CONFIG, zkUrl);
        conf.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        conf.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        streamsConfig = new StreamsConfig(conf);
    }
    public void run(){
        KafkaStreams streams = new KafkaStreams(kStreamBuilder, streamsConfig);
        streams.start();
    }

    public static void main(String[] args) {
        DauTopology dauTopology = new DauTopology();
        String kafkaServer = "localhost:9092";
        String zkServer = "localhost:2181";
        dauTopology.buildTopology(Constants.DAU_COUNT_TOPIC,Constants.DAU_SOURCE_TOPIC);

        dauTopology.setConfig(kafkaServer, zkServer);
        try {
            dauTopology.run();
            LOG.info("kafka broker list is:" +kafkaServer);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
